package com.etl.component.file;

import cn.hutool.core.io.FileUtil;
import cn.hutool.core.util.IdUtil;
import cn.hutool.core.util.StrUtil;
import com.google.common.collect.Lists;
import com.component.api.constants.ComponentCodeConstants;
import com.component.api.constants.ComponentConstants;
import com.component.api.excepiton.ComponentException;
import com.component.api.excepiton.FlowRunningException;
import com.component.api.model.PortData;
import com.component.api.model.ProcessResult;
import com.component.api.model.data.ColumnData;
import com.component.api.model.data.SetData;
import com.component.api.model.dto.FileCenterConfigDTO;
import com.component.api.model.param.TableColumn;
import com.component.api.model.param.ParamPair;
import com.common.utils.constant.Constants;
import com.etl.component.common.AbstractFlinkComponent;
import com.etl.component.file.center.FileCenter;
import com.etl.component.file.center.FileCenterManager;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.io.RowCsvInputFormat;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.typeutils.RowTypeInfo;
import org.apache.flink.core.fs.Path;
import org.apache.flink.types.Row;
import org.jetbrains.annotations.NotNull;

import java.io.*;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;

/**
 * 描述：
 *  文件读取都是读出来都是字符串类型的
 * @author xianggj
 * @Date 2021/10/12 18:10
 **/
public class CsvReadComponent extends AbstractFlinkComponent {

    @Override
    public ProcessResult process(List<PortData> datas, List<ParamPair> paramPairs) {
        Map<String, Object> params = paramToObject(paramPairs);
        //文件中心路径
        String path = (String)params.get(ComponentConstants.PATH);
        String type = (String)params.get(ComponentConstants.TYPE);
        InputStream inputStream = null;
        try {
            FileCenterConfigDTO config = (FileCenterConfigDTO)params.get(ComponentConstants.CONFIG);

            inputStream = FileCenterManager.getFileCenter(FileCenter.MINIO)
                    .getFileStream(config, path);
        } catch (Exception e) {
            e.printStackTrace();
            throw new ComponentException("文件读取失败", e);
        }
        //临时文件
        File dir = new File(FileCenter.CHACE_PATH);
        FileUtil.mkdir(dir);
        File file = FileUtil.writeFromStream(inputStream, dir + File.separator +
                IdUtil.fastSimpleUUID() + Constants.DOT + ComponentConstants.CSV);
        DataSet<Row>  input = null;
        List<String> headers = (List<String>) params.get(ComponentConstants.HEADER);
        List<TableColumn> columns = getTableColumns(headers);
        if (ComponentConstants.CSV.equalsIgnoreCase(type)){
            final ExecutionEnvironment env = getEnv();
            String encoding = (String)params.getOrDefault(ComponentConstants.ENCODING, "UTF-8");
            Boolean isFirstHeader = (Boolean) params.get(ComponentConstants.IS_FIRST_LINE);
            boolean isHeader = isFirstHeader == null? false: isFirstHeader;
            TypeInformation[] fieldTypeInfos = new TypeInformation[headers.size()];
            Arrays.fill(fieldTypeInfos, BasicTypeInfo.STRING_TYPE_INFO);
            RowCsvInputFormat rowCsvInputFormat =
                    new RowCsvInputFormat(new Path(file.toURI()), fieldTypeInfos);
            //设置编码
            rowCsvInputFormat.setCharset(encoding);
            //跳过第一行
            rowCsvInputFormat.setSkipFirstLineAsHeader(isHeader);
            //不规则csv解析
            rowCsvInputFormat.setLenient(true);
            //启用引用字符串解析 当字符串中间有分隔符逗号“,”使用引用做特征字符串转义
            rowCsvInputFormat.enableQuotedStringParsing('\"');
            RowTypeInfo rowTypeInfo = new RowTypeInfo(fieldTypeInfos, headers.toArray(new String[0]));
            input = env.createInput(rowCsvInputFormat, rowTypeInfo);

        } else {
            throw new ComponentException("文件格式错误");

        }
        return execute(input, columns,"执行csv拉取操作");
    }

    @Override
    public List<PortData> previewOutputColumns(List<PortData> datas, List<ParamPair> paramPairs) {
        Map<String, Object> params = paramToObject(paramPairs);
        List<String> headers = (List<String>) params.get(ComponentConstants.HEADER);
        List<TableColumn> columns = getTableColumns(headers);
        return Lists.newArrayList(new PortData(getRule(), new ColumnData(columns)));
    }

    @NotNull
    private List<TableColumn> getTableColumns(List<String> headers) {
        List<TableColumn> columns = new ArrayList<>();
        for (int i = 0; i < headers.size(); i++) {
            String header = headers.get(i);
            columns.add(new TableColumn(header, ComponentConstants.VARCHAR, ComponentConstants.DEFAULT_VARCHAR_LENGTH));
        }
        return columns;
    }

    @Override
    public String getRule() {
        return ComponentCodeConstants.CSV_PARSING;
    }
}
